package org.apache.zookeeper.server.paxos2;

import java.util.*;
import java.util.concurrent.*;

public class DataCache {

    private Storage storage = null;

    private class CacheLine {
        public long blockId = -1;
        public byte[] data;
        public long version;
    }

    private LinkedBlockingQueue<CacheLine> cleanList = new LinkedBlockingQueue<CacheLine>(Configuration.cacheSize);

    private LinkedBlockingQueue<CacheLine> dirtyList = new LinkedBlockingQueue<CacheLine>(Configuration.cacheSize);

    private HashMap<Long, CacheLine> maps = new HashMap<Long, CacheLine>();

    private FlushThread flushThread = new FlushThread();

    public DataCache(Storage storage) {
        this.storage = storage;
        try {
            for (int i = 0; i < Configuration.cacheSize; i++)
                cleanList.put(new CacheLine());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        flushThread.start();
    }

    private void replace(CacheLine c, long blockId, byte[] data, long version) {
        if (c.blockId != -1)
            maps.remove(c.blockId);
        c.blockId = blockId;
        c.data = data;
        c.version = version;
        maps.put(c.blockId, c);
    }

    public void write(long blockId, byte[] data, long version) {
        /*if(true){
          storage.writeToStripe(blockId, data);
          return;
      }*/
        try {
            CacheLine c = cleanList.take();
            replace(c, blockId, data, version);
            dirtyList.put(c);
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public byte[] read(long blockId) {
        return null;
        /*CacheLine c = maps.get(blockId);
        if (c != null) {
            //System.out.println("read from cache "+blockId);
            return c.data;
        } else {
            byte[] data = storage.readFromStripe(blockId);
            long version = -1; //need fix
            try {
                c = cleanList.take();
                replace(c, blockId, data, version);
                cleanList.put(c);
                return c.data;
            }
            catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        }        */
    }

    public void close() {
        flushThread.interrupt();
    }

    private class FlushThread extends Thread {

        public void run() {
            LinkedList<CacheLine> cleaned = new LinkedList<CacheLine>();
            while (true) {
                try {
                    CacheLine c = dirtyList.poll(5, TimeUnit.MILLISECONDS);
                    if (c != null) {
                        storage.writeToStripe(c.blockId, c.data);
                        cleaned.add(c);
                    }

                    if (cleaned.size() > 10000 || (c == null && cleaned.size() > 0)) {
                        //	storage.flushStripes();
                        for (CacheLine tmp : cleaned)
                            cleanList.put(tmp);
                        cleaned.clear();
                    }
                } catch (InterruptedException e) {
                    System.out.println("FlushThread exited");
                    return;
                    //e.printStackTrace();
                }
            }
        }
    }

}
